import asyncio
import akshare as ak
import asyncio.queues as Queue
import pandas as pd


class Strategy:
    sma_period = 5  # 均线周期

    def __init__(self, data):
        self.data = data

    def avg_Signal(self, data):
        if len(data) < self.sma_period:
            return
        data['sma'] = data['LastPrice'].rolling(window=self.sma_period).mean()
        print('最近5条K线的均价 %s ,当前收盘价 %s' % (data.tail(1)['sma'], data.tail(1)['LastPrice']))
        if data['LastPrice'] > data['sma']:
            # 如果收盘价超过移动平均线，买入
            print(f'收盘价超过移动平均线， 买入')
        else:
            # 如果收盘价低于移动平均线，卖出
            print(f'如果收盘价低于移动平均线， 卖出')


class HandlerData:
    sma_period = 5  # 均线周期

    def __init__(self, queue, data):
        self.queue = queue
        self.data = data

    async def process_queue(self):
        while True:
            datas = await self.queue.get()
            if datas is None:
                # Signal to stop processing
                break
            print('HandlerData datas===', datas)
            self.data = pd.concat([self.data, datas], axis=0)
            print('HandlerData data.size()===', len(self.data))


class QueueManager():
    def __init__(self, queue):
        self.queue = queue

    def enqueue_data(self, data):
        self.queue.put_nowait(data)


class QueryData:
    def __init__(self, queue: QueueManager, symbol, num_init_backfill_bars):
        self.queue = queue
        self.symbol = symbol
        self.num_init_backfill_bars = num_init_backfill_bars

    # 获取分时行情数据并写入到CSV
    def futures_zh_minute_sina_to_csv_bt(self):
        futures_sina_df = ak.futures_zh_minute_sina(symbol=self.symbol, period="1")
        futures_sina_df = futures_sina_df.tail(self.num_init_backfill_bars)
        # 改列名
        futures_sina_df.columns = ['datetime', 'OpenPrice', 'HighPrice', 'LowPrice', 'LastPrice', 'BarVolume', 'hold']
        # 增加symbol列
        futures_sina_df['local_symbol'] = self.symbol + '.SHFE'
        futures_sina_df['OpenInterest'] = 0
        futures_sina_df.index = futures_sina_df['datetime']

        # pd.concat([data, futures_sina_df], axis=0)

        self.queue.enqueue_data(futures_sina_df)
        self.queue.enqueue_data(None)


class StrategyHandler:
    def __init__(self, data, queueManager: QueueManager, num_init_backfill_bars=100):
        self.data = data
        self.queue = queueManager.queue
        self.num_init_backfill_bars = num_init_backfill_bars
        self.is_live = False
        self.queueManager = queueManager

    def change_live(self, bool=True):
        self.is_live = bool
        self.num_init_backfill_bars=1
        pass

    async def strategy_start(self):
        print('strategy_start')
        print(self.data)


async def my_async_function():
    queue = Queue.Queue()
    queue_manager = QueueManager(queue)
    data = pd.DataFrame()
    strategyHandler = StrategyHandler(data, queue_manager, 100)
    handlerData = HandlerData(queue, data)
    strategy = Strategy(data)
    queryData = QueryData(queue_manager, 'rb2401', strategyHandler.num_init_backfill_bars)
    while True:
        queryData.num_init_backfill_bars = strategyHandler.num_init_backfill_bars
        queryData.futures_zh_minute_sina_to_csv_bt()
        await handlerData.process_queue()
        strategy.avg_Signal(handlerData.data)
        strategyHandler.change_live(True)
        await asyncio.sleep(60)



if __name__ == "__main__":
    # 创建事件循环
    loop = asyncio.get_event_loop()

    try:
        # 运行异步函数
        loop.run_until_complete(my_async_function())
    except KeyboardInterrupt:
        # 处理键盘中断，例如按下 Ctrl+C
        pass
    finally:
        # 关闭事件循环
        loop.close()
